//
// file : inet_ipc.c
// date : 06/03/07
//
#include <string.h>
#include <stdlib.h>
#include <ctype.h>
#include "inet_ipc.h"
#include "inet_tmr.h"

#define MAKE_SESS_ID_( didx, slotid )  (long)(((didx)<<16)|(slotid&0xffff))
#define GET_SLOT_ID__( sid )           (long)(((unsigned long)sid)&0xffff)
#define GET_DOMAIN_ID( sid )           (long)(((unsigned long)(sid))>>16)
//
// in smp systems a domain is associate with an execution thread.
//
#ifndef MAX_IPC_DOMAINS
    #define  MAX_IPC_DOMAINS 64
#endif
static ipc_domain_t * g_domain[MAX_IPC_DOMAINS];

///////////////////////////////////////////////////////////////////////////////
//
//  internal functions
//
///////////////////////////////////////////////////////////////////////////////

//
//
//
void  ipc_reset__(ipccb_t *ipc)
{
    // keep cbsize and slotid

    assert( ipc->cbsize >= sizeof(*ipc) );

    // set zero all struct but not extra
    memset( ((char*)ipc) + sizeof(ipc->cbsize), 
            0, 
            sizeof(*ipc) - sizeof(ipc->cbsize) );
    /*
    memset( ipc->nuance, 0, sizeof(ipc->nuance) );
    */
    ipc->socket = INVALID_SOCKET;
    ipc->sotype = 0;

    ipc->apcsem = (void*)~0;
    /*
    memset( &ipc->loaddr, 0, sizeof(ipc->loaddr) );
    memset( &ipc->toaddr, 0, sizeof(ipc->toaddr) );
    memset( &ipc->mcaddr, 0, sizeof(ipc->mcaddr) );

    ipc->sflags = 0;
    ipc->tagain = 0;

    ipc->states = 0;

    ipc->obsize = 0;
    ipc->ocount = 0;
    ipc->obound = 0;
    ipc->ibsize = 0;
    ipc->ipoint = 0;
    ipc->ibound = 0;

    ipc->reader = 0;
    ipc->writer = 0;
    ipc->closed = 0;
    ipc->tready = 0;
    ipc->peeker = 0;

    ipc->uflags = 0;

    memset( ipc->usrobj, 0, sizeof(ipc->usrobj) );
    memset( ipc->uparam, 0, sizeof(ipc->uparam) );
    memset( ipc->ustate, 0, sizeof(ipc->ustate) );
    memset( ipc->cbtext, 0, sizeof(ipc->peerid) );
    */
}


//
// allocate ipc context
//
ipccb_t * ipc_alloc__(ulong extra)
{
    ipccb_t *ipc = (ipccb_t*) malloc( sizeof(*ipc) + extra );

    if( ipc )
    {
        ipc->cbsize = sizeof(*ipc) + extra;

        ipc_reset__( ipc );
    }

    return ipc;
}


//
// addr_str must be in the format "ddd.ddd.ddd.ddd:port"
//
long set_inaddr(char *addr_str, struct in_addr *in)
{
    char ip[16];
    int  k;

    if( addr_str == 0 || in == 0 ) return -1001;

    for( k = 0; addr_str[k]=='.' || isdigit(addr_str[k]); k++ )
    {
        if( k >= 15 ) return -1002;

        ip[k] = addr_str[k];
    }

    ip[k] = '\0';

    in->s_addr = inet_addr(ip);

    return k;
}

long set_soaddr(char *addr_str, struct sockaddr_in *sin)
{
    long  k    = 0;
    long  port = 0;

    if( 0 == addr_str || 0 == sin ) return 0;

    memset( sin, 0, sizeof(*sin) );

    // addr_str is in the form of <ip>:<port>
    if( strchr( addr_str, '.' )  )
    {
        k = set_inaddr( addr_str, &(sin->sin_addr) );
    }
    else
    {
        k = 0;  // don't have ip address
    }

    if( k < 0 ) return k;

    // ':' is optional when no ip address is present
    if( addr_str[k] == ':' ) k += 1;

    if( addr_str[k] != '\0' )
    {
        if( ! isdigit(addr_str[k]) ) return -1003;

        port = atoi( addr_str + k );

        if( port <= 0 || port >= (1<<16) ) return (-1004);
        sin->sin_port   = htons((unsigned short)port);
    }

    sin->sin_family = AF_INET;

    return port;
}

//
// join multicast group
//
int  join_mgroup(socket_t msock, struct in_addr *mc_addr, struct in_addr *lo_addr)
{
    // join a multicast group
    int            e, v;

    // set multicast TTL 
    v = 8;

    e = setsockopt(msock, IPPROTO_IP, IP_MULTICAST_TTL, (char*)&v, sizeof(v) );

    if( e )
    {
        ipc_print("join_mgroup set IP_MULTICAST_TTL error(%d)\n",
            sock_error());
    }

    // multicast loopback
    v = 0;  

    e = setsockopt(msock, IPPROTO_IP, IP_MULTICAST_LOOP, (char*)&v, sizeof(v));

    if( e ) 
    {
        ipc_print("join_mgroup set IP_MULTICAST_LOOP error(%d)\n",
            sock_error());
    }

    // set multicast interface
    e = setsockopt(msock, IPPROTO_IP, IP_MULTICAST_IF, 
        (char*)lo_addr, sizeof(*lo_addr));

    if( e )
    {
        ipc_print("join_mgroup set IP_MULTICAST_IF error(%d)\n",
            sock_error());
    }

#if defined( INETIPC_USE_IGMPV3 )
    {
        ip_mreq_source imreq;  // igmp v3

        memset(&imreq, 0, sizeof(imreq));
        imreq.imr_multiaddr  = *mc_addr;
        imreq.imr_sourceaddr = INADDR_ANY;
        imreq.imr_interface  = *lo_addr;

        e = setsockopt(msock, IPPROTO_IP, IP_ADD_SOURCE_MEMBERSHIP, 
            (char*)&imreq, sizeof(imreq));  
    }
    if( e )
    {
        ipc_print("join_mgroup set IP_ADD_SOURCE_MEMBERSHIP error(%d)\n",
            sock_error());
    }

#else // INETIPC_USE_IGMPV3
    {
        struct ip_mreq    imreq;

        memset(&imreq, 0, sizeof(imreq));
        imreq.imr_multiaddr = *mc_addr;
        imreq.imr_interface = *lo_addr;

        e = setsockopt(msock, IPPROTO_IP, IP_ADD_MEMBERSHIP, 
            (char*)&imreq, sizeof(imreq));  
    }
    if( e )
    {
        ipc_print("join_mgroup set IP_ADD_MEMBERSHIP error(%d)\n",
            sock_error() );
    }
#endif // INETIPC_USE_IGMPV3


    //ipc_print("join_mgroup(%s) done\n", inet_ntoa(imreq.imr_multiaddr));

    return e;
}

int  drop_mgroup(socket_t msock, struct in_addr *mgroup, struct in_addr *lo_addr)
{
    int            e;

#if defined( INETIPC_USE_IGMPV3 )

    ip_mreq_source imreq;  // igmp v3

    memset(&imreq, 0, sizeof(imreq));
    imreq.imr_multiaddr  = to_addr->sin_addr;
    imreq.imr_sourceaddr = lo_addr->sin_addr; 
    imreq.imr_interface  = to_addr->sin_addr;

    e = setsockopt(msock, IPPROTO_IP, IP_DROP_SOURCE_MEMBERSHIP, 
        (char*)&imreq, sizeof(imreq));  

    if( e )
    {
        ipc_print("drop_mgroup() set IP_DROP_SOURCE_MEMBERSHIP error(%d)\n", sock_error());
    }

#else  // INETIPC_USE_IGMPV3

    struct ip_mreq    imreq;

    memset(&imreq, 0, sizeof(imreq));
    imreq.imr_multiaddr = *mgroup;
    imreq.imr_interface = *lo_addr;

    e = setsockopt(msock, IPPROTO_IP, IP_DROP_MEMBERSHIP, 
        (char*)&imreq, sizeof(imreq));  

    if( e )
    {
        ipc_print("drop_mgroup set IP_DROP_MEMBERSHIP error(%d)\n", sock_error());
    }
#endif // INETIPC_USE_IGMPV3

    return e;
}


///////////////////////////////////////////////////////////////////////////////
//
//  exported functions
//
///////////////////////////////////////////////////////////////////////////////


//
// get domain by index
//
void * __stdcall get_domain_(long didx)
{
    if( didx <= 0 || didx > MAX_IPC_DOMAINS )
    {
        return ipc_domain_( 0 ); // search by thread
    }

    // index is 1-based
    if( g_domain[ didx-1 ] )
    {
        assert( g_domain[didx-1]->didx == didx );
    }

    return g_domain[didx-1];
}

//
// search for ipc domain by session id
//
void * __stdcall ipc_domain_( long sid )
{
    long          didx, k;
    ulong         thid = thread_id_();
    ipc_domain_t *domain;

    didx = GET_DOMAIN_ID( sid );

    // if we get a good domain index
    if( didx > 0 && didx <= MAX_IPC_DOMAINS ) 
    {
        domain = get_domain_( didx );

        return domain;
    }

    // didx == 0, search by thread id
    // must be in-thread

    for( k = 0; k < MAX_IPC_DOMAINS; k++ )
    {
        ipc_domain_t *domain = g_domain[ k ];

        if( domain && domain->thid == thid )
        {
            // index is 1-based
            assert( k+1 == domain->didx );

            return domain;
        }
    }

    // we may get here if we search for domain out of thread

    return 0;
}

//
// allocate inet ipc domain context - called in-thread
//
long  __stdcall ipc_initial(long max_sess, long ms_beat)
{
    ipc_domain_t *domain;
    long          k;

    size_t        d_size;
    size_t        q_size;

    domain = ipc_domain_( 0 );  // find domain by thread

    if( 0 != domain )
    {
        return domain->didx;  // exists
    }

    // create new

    for( k = 0; k < MAX_IPC_DOMAINS; k++ )
    {
        if( g_domain[k] != 0 ) continue;

        if( max_sess <= 16 ) max_sess = IPCS_PER_DOMAIN;
        if( max_sess < 1   ) max_sess = 1;  // guard against bad IPCS_PER_DOMAIN

        // estimate timer queue memory
        tmr_alloca_(0, &q_size, TIMER_QUEUE_SIZE); 

        d_size = sizeof( ipc_domain_t ) + ( max_sess - 1 ) * sizeof( ipccb_t* );

        d_size = (d_size+15)/16*16;

        domain = (ipc_domain_t*)malloc( d_size + q_size );

        memset( domain, 0, d_size );

        if( domain == 0 ) return 0; // memory overflow

        domain->qtmr = tmr_alloca_((char*)domain + d_size, &q_size, TIMER_QUEUE_SIZE);

        // set owner hook
        tmr_setownr(domain->qtmr, domain);

        domain->didx = k + 1;  // domain index starting at 1

        domain->ncbs = max_sess;

        domain->thid = thread_id_();  // 

        g_domain[k] = domain;

        break;  // done
    }

    if( k >= MAX_IPC_DOMAINS )
    {
        ipc_print(" ipc_initial() exceeds max number of ipc domains(%d)\n", k);

        k = -1;
    }
    else
    {
        // platform specific initialization
        ipc_osinit_( domain );
    }

    // start timer heart beat
    if( ms_beat > 0 ) 
    {
        tmr_beating( domain->qtmr, ms_beat);
    }
    else if( ms_beat < 0 )
    {
        // using default beat rate
        tmr_beating( domain->qtmr, 0);
    }

    return domain->didx;
}

//
// invalidate ipc domain
//
void  __stdcall ipc_cleanup(long didx)
{
    ipc_domain_t *domain;

    if( didx <= 0 || didx > MAX_IPC_DOMAINS ) return;

    domain = get_domain_( didx );

    assert( didx == domain->didx );

    // index is 1-based
    assert( domain == g_domain[didx - 1] );

    // note: we are out-of-thread here. 
    //       leave cleanup to in-thread routines

    // terminate thread
    ipc_osexit_( domain );

    free( g_domain[didx - 1] );

    g_domain[didx - 1] = 0;
}

//
// return a session id
//  session id consists of domain index and slot id
//
long  __stdcall ipc_session(long didx, ulong extra)
{
    long          k;
    ipc_domain_t *domain = get_domain_( didx );

    // didx could be zero for in-thread calls
    didx = domain->didx;  

    if( domain == 0 ) return INVALID_SESSION_;

    for( k = 0; k < domain->ncbs; k++ )
    {
        ipccb_t *ipc = domain->ipcs[k];

        if( ipc == 0 ) 
        {
            domain->ipcs[k] = ipc_alloc__( extra );
            domain->ipcs[k]->sessid = MAKE_SESS_ID_(didx, k);
            domain->ipcs[k]->states = IPC_STAT_SESSION;

            return domain->ipcs[k]->sessid;
        }

        // match ipc block size
        if( (ipc->states == IPC_STAT_INVALID) && (ipc->cbsize == extra + sizeof(*ipc)) ) 
        {
            ipc_reset__( ipc );

            ipc->states = IPC_STAT_SESSION;

            ipc->sessid = MAKE_SESS_ID_(didx, k);

            return ipc->sessid;   // reuse
        }
    }

    assert(0); // memory 

    return INVALID_SESSION_;
}

//
//  return ipccb_t block. if sid is a slotid, this is in-thread call
//
void* __stdcall ipc_context(long sid)
{
    long          slotid;
    ipc_domain_t *domain;

    domain = ipc_domain_( sid ); // if didx is zero, this is in-thread call

    if( domain == 0 ) return 0;

    slotid = GET_SLOT_ID__( sid );

    if( slotid >= domain->ncbs || domain->ipcs[slotid] == 0 )
    {
        ipc_print(" ipc_context(sid=%d) invalid slot id\n", sid);
        return 0;
    }

    assert( slotid == GET_SLOT_ID__(domain->ipcs[slotid]->sessid) );

    return domain->ipcs[slotid];
}

long  __stdcall set_freader(long sid, f_reader f)
{
    ipccb_t *ipc = ipc_context(sid);

    if( ipc == 0 ) return -1030;

    ipc->reader = f;

    return 0;
}

long  __stdcall set_fwriter(long sid, f_callup f)
{
    ipccb_t *ipc = ipc_context(sid);

    if( ipc == 0 ) return -1040;

    ipc->writer = f;

    return 0;
}

long  __stdcall set_fpeeker(long sid, f_callup f)
{
    ipccb_t *ipc = ipc_context(sid);

    if( ipc == 0 ) return -1050;

    ipc->peeker = f;

    return 0;
}
long  __stdcall set_fclosed(long sid, f_callup f)
{
    ipccb_t *ipc = ipc_context(sid);

    if( ipc == 0 ) return -1060;

    ipc->closed = f;

    return 0;
}

long  __stdcall set_ftready(long sid, f_callup f)
{
    ipccb_t *ipc = ipc_context(sid);

    if( ipc == 0 ) return -1070;

    ipc->tready = f;

    return 0;
}

long  __stdcall set_userobj(long sid, int index, void *obj)
{
    ipccb_t *ipc = ipc_context(sid);

    if( ipc == 0 ) return -1080;

    if( index >= NUM_USER_OBJECTS )
    {
        return -1082;
    }

    ipc->usrobj[index] = obj;

    return 0;
}

void* __stdcall get_userobj(long sid, int index)
{
    ipccb_t *ipc = ipc_context(sid);

    if( ipc == 0 ) return 0;

    if( index >= NUM_USER_OBJECTS )
    {
        assert(0);
        return 0;
    }

    return ipc->usrobj[index];
}

//
// xet_u_flags() can be called from outside of the thread
//
ulong __stdcall xet_u_flags(long sid, ulong flags)
{
    ipccb_t *ipc = ipc_context(sid);

    if( ipc == 0 ) return 0;

    ipc->uflags ^= flags;

    return ipc->uflags;
}

ulong __stdcall xet_s_flags(long sid, ulong flags)
{
    ipccb_t *ipc = ipc_context(sid);

    if( ipc == 0 ) return 0;

    ipc->sflags ^= flags;

    return ipc->sflags;
}

//
// set local socket address
//
long  __stdcall set_lo_addr(long sid, char *lo_addr_str)
{
    ipccb_t *ipc = ipc_context(sid);

    if( ipc == 0 ) return -1090;

    return set_soaddr( lo_addr_str, &(ipc->loaddr) );
}

//
// set peer socket address
//
long  __stdcall set_to_addr(long sid, char *to_addr_str)
{
    ipccb_t *ipc = ipc_context(sid);

    if( ipc == 0 ) return -1100;

    return set_soaddr( to_addr_str, &(ipc->toaddr) );
}

long  __stdcall set_mc_addr(long sid, char *mc_addr_str)
{
    ipccb_t *ipc = ipc_context(sid);

    if( ipc == 0 ) return -1110;

    return set_inaddr( mc_addr_str, &(ipc->mcaddr) );
}

//
// set tcp retry interval
//
long  __stdcall set_t_again(long sid, long interval)
{
    ipccb_t *ipc = ipc_context(sid);

    if( ipc == 0 ) return -1120;

    ipc->tagain = interval; //

    return 0;
}

///////////////////////////////////////////////////////////////////////////////
//
// can be called from out-of-thread. to handle errors
// in creating tcp/udp sessions
//
void* __stdcall ipc_reclaim(long sid)
{
    ipccb_t *ipc;

    ipc = ipc_context( sid );

    if( ipc == 0 ) return 0;

    if( ipc->socket != INVALID_SOCKET )
    {
        sio_close( ipc->socket );

        ipc->socket = INVALID_SOCKET;
    }

    ipc->states = IPC_STAT_INVALID;

    return ipc;
}

//
// shutdown ipc session. 
//
long  __stdcall ipc_shutoff(long sid)
{
    ipc_domain_t *domain;
    ipccb_t      *ipc;
    long          ret;

    ipc = ipc_context(sid);

    if( ipc == 0 ) return (-1010);

    // check thread id
    domain = ipc_domain_( sid );

    if( domain == 0 ) return (-1011);

    if( domain->thid != thread_id_() )
    {
        // out-thread call, thunk
        ipc->apcfun = ipc_shutoff;

        ipc->apcsem = (void*) ~0;  // wait for return value

        ret = ipc_waitapc( domain, ipc );
    }
    else
    {
        // in thread call
        if( ipc->socket != INVALID_SOCKET )
        {
            sio_close( ipc->socket );

            ipc->socket = INVALID_SOCKET;
        }

        ipc->states = IPC_STAT_INVALID;

        ret = 0;
    }
    return ret;
}

//
// write a block of data to socket. if 'len' is negative, don't
// copy leftover to outbound buffer
// caller should close session if getting return of -1
//
long  __stdcall ipc_writing(long sid, void *data, int len)
{
    ipccb_t      *ipc = ipc_context(sid);
    ipc_domain_t *domain;

    if( ipc == 0 ) return -1015; // 

    // check thread id
    domain = ipc_domain_( sid );

    if( domain == 0 ) return (-1016);

    if( domain->thid != thread_id_() )
    {
        long  ret;
        // out-thread call, thunk
        ipc->apcfun    = ipc_writing;
        ipc->apcarg[0] = data;
        ipc->apcarg[1] = (void*)len;
        ipc->apcsem    = (void*)~0; // wait

        ret = ipc_waitapc(domain, ipc);

        return ret;
    }

    // send in buffer outbound data 
    if( ipc->ocount > 0 && ipc->socket != INVALID_SOCKET )
    {
        ipc_sending( ipc );
    }

    // if len < 0, we don't queue data to outbound buffer
    if( len != 0 && data != 0)
    {
        long k = (len > 0 ? len : - len);  // number of bytes to send
        
        if( ipc->ocount == 0 && ipc->socket != INVALID_SOCKET )
        {
            if( ipc->states & IPC_STAT_CONNECT )
            {
                k = send(ipc->socket, data, k, 0);
            }
            else 
            {
                int slen = sizeof( ipc->toaddr );

                k = sendto(ipc->socket, data, k, 0,
                           (struct sockaddr*)&ipc->toaddr, slen );
            }

            if( k < 0 && sock_error() == EWOULDBLOCK ) k = 0;
        }

        // if len is positive, copy the rest to outbound buffe
        if( k >= 0 && len > 0 )
        {
            len -= k;  // leftover

            if( len > 0 )
            {
                if( len > ipc->obsize - ipc->ocount )
                {
                    len = ipc->obsize - ipc->ocount;
                }

                // queue data to outbound buffer
                memmove( ipc->obound + ipc->ocount, (char*)data + k, len);
            }

            ipc->ocount += len;

            len += k;  // number of bytes taken
        }
        else 
        {
            // either socket error or number of bytes sent (no-queueing)
            len = k; 
        }
    }

    // return negative for error or number of bytes taken/sent
    return len; 
}

//
// caller should have set reader and/or writer before calling.
// a multicast group can be specified as local address or
// in the form of <mgroup>@<local-if>:<port>
//
long  __stdcall udp_binding(long sid, char *addr_str)
{
    int      e;
    ipccb_t *ipc = ipc_context(sid);

    if( ipc == 0 ) return -1120;

    if( ipc->socket == INVALID_SOCKET ) return -1130;

    if( addr_str )
    {
        // a multicast group can be specified in local address in the form
        // "<mgroup>@<local-if>"
        char *p = strchr( addr_str, '@' );

        if( p != 0 )  // a multicast group is given
        {
            set_inaddr( addr_str, &ipc->mcaddr );

            if( ! IS_MULTICAST( ipc->mcaddr.s_addr ) )
            {
                memset(&ipc->mcaddr, 0, sizeof(ipc->mcaddr)); 

                return -1140;
            }

            p += 1;  // points to local address
        }
        else
        {
            p = addr_str;
        }

        if( p && *p )
        {
            set_soaddr(p, &ipc->loaddr);
        }
    }

    // if local adress is a multicast group
    if( IS_MULTICAST( ipc->loaddr.sin_addr.s_addr ) )
    {
        ipc->mcaddr = ipc->loaddr.sin_addr;

        ipc->loaddr.sin_addr.s_addr = INADDR_ANY;  // overide 
    }

    e = bind( ipc->socket, (struct sockaddr*)&ipc->loaddr, sizeof(ipc->loaddr));

    if( e < 0 )
    {
        return -1160;
    }

    // if multicast group is defined
    if( IS_MULTICAST( ipc->mcaddr.s_addr ) )
    {
        // in windows do this after bind()
        e = join_mgroup(ipc->socket, &ipc->mcaddr, &ipc->loaddr.sin_addr);

        if( e < 0 )
        {
            return -1150;
        }
    }

    // set SOC_FLAG_PASSIVE
    if( 0 == ipc->toaddr.sin_addr.s_addr )
    {
        ipc->sflags |= SOC_FLAG_PASSIVE;
    }

    if( 0 != ipc->toaddr.sin_addr.s_addr )
    {
        // for udp socket we do not expect a socket event due
        e = connect( ipc->socket, 
                     (struct sockaddr*)&ipc->toaddr, 
                     sizeof(ipc->toaddr) );

        if( e < 0 ) return -1170;

        ipc->states |= IPC_STAT_CONNECT;
    }

    return e;
}

//
// caller should have set reader and/or writer before this call.
// no need to join multicast group to send to a multicats group. 
// application should use a pair of ports to make to make in-host
// multicast communications possible
//
long  __stdcall udp_connect(long sid, char *to_addr_str)
{
    int      e;
    ipccb_t *ipc;

    ipc = ipc_context(sid);

    if( ipc == 0 ) return -1180;

    if( to_addr_str )
    {
        if( set_soaddr(to_addr_str, &ipc->toaddr) < 0 )
        {
            // error
            ipc_print("udp_connect() error addr string(%s)\n", to_addr_str);

            return -1190;
        }
    }

    // for udp socket we do not expect a socket event due
    e = connect( ipc->socket, 
                 (struct sockaddr*)&ipc->toaddr, sizeof(ipc->toaddr) );

    if( e < 0 ) return -1200;

    ipc->states |= IPC_STAT_CONNECT;

    return 0;
}

//
// create an ipc session context and open a udp socket
//
long  __stdcall udp_session( long didx )
{
    long     sid;
    ipccb_t *ipc;

    sid = ipc_session( didx, IPC_UDP_IBUF_SIZE + IPC_UDP_OBUF_SIZE );

    if( sid == INVALID_SESSION_ )
    {
        ipc_print(" udp_session() ipc_session(sid=%d)\n", sid);
        return -1210;
    }

    ipc = ipc_context( sid );

    if( ipc == 0 ) return -1211;

    ipc->ibsize = IPC_UDP_IBUF_SIZE;
    ipc->ibound = (char*)(ipc+1);
    ipc->obsize = IPC_UDP_OBUF_SIZE;
    ipc->obound = ipc->ibound + ipc->ibsize;

    ipc->sotype = SOCK_DGRAM;

    ipc->socket = socket(AF_INET, ipc->sotype, 0);

    if( ipc->socket == INVALID_SOCKET )
    {
        ipc_print(" udp_session() error=%d\n", sock_error());
    }

    // socket events hookup
    ipc_hookup_( ipc_domain_(sid), ipc );

    return sid;
}


//
// caller should have set peer address
//
long  __stdcall tcp_connect(long sid, char *to_addr_str)
{
    int      e;
    ipccb_t *ipc = ipc_context(sid);

    if( ipc == 0 ) return -1220;

    if( ipc->socket == INVALID_SOCKET ) return -1230;

    if( to_addr_str )
    {
        set_soaddr( to_addr_str, &ipc->toaddr );
    }

    if( ipc->toaddr.sin_addr.s_addr == INADDR_ANY )
    {
        return -1240;
    }

    ipc->sflags &= (~SOC_FLAG_PASSIVE);

    // if local address is given, bind to it
    if( ipc->loaddr.sin_port != 0 || ipc->loaddr.sin_addr.s_addr != INADDR_ANY )
    {
        e = bind( ipc->socket, (struct sockaddr*)&ipc->loaddr, sizeof(ipc->loaddr));

        if( e < 0 )
        {
           ipc_print(" tcp_connect(sid=%d) bind()=%d\n", sid, sock_error());
           return -1250;
        }
    }

    e = connect( ipc->socket, (struct sockaddr*)&ipc->toaddr, sizeof(ipc->toaddr) );

    if( e < 0 && (sock_error() != EINPROGRESS) ) 
    {
        ipc_print(" tcp_connect(sid=%d) connect()=%d\n", sid, sock_error());
        return -1260;
    }
    else
    {
        e = 0;
    }

    return e;
}

//
// http client function
//
long  __stdcall tcp_request(long sid, char *req, long len)
{
    int   e;
    ipccb_t *ipc = ipc_context(sid);

    if( ipc == 0 ) return -1270;

    if( len > ipc->obsize ) return -1280;

    if( req && len > 0 )
    {
        // copy outbound data
        memcpy( ipc->obound, req, len );

        ipc->ocount = len;
    }

    // peer address must have set
    e = tcp_connect( sid, 0 );

    return e;
}

//
// prepare a tcp context block
//
ipccb_t *tcp_context( long didx, long ibsize, long obsize )
{
    long     sid;
    ipccb_t *ipc;

    if( ibsize < 0 ) ibsize = IPC_TCP_IBUF_SIZE;
    if( obsize < 0 ) obsize = IPC_TCP_OBUF_SIZE;

    sid = ipc_session( didx, ibsize + obsize );

    if( sid == INVALID_SESSION_ )
    {
        ipc_print(" tcp_context() ipc_session(sid=%d)\n", sid);
        return 0;
    }

    ipc = ipc_context( sid );

    if( ipc == 0 )
    {
        return 0;
    }
    //ipc_print(" tcp_context(sid=%d sessid=%d)\n", sid, ipc->sessid);

    ipc->ibsize = ibsize;
    if( ibsize > 0 ) ipc->ibound = (char*)(ipc+1);

    ipc->obsize = obsize;
    if( obsize > 0 ) ipc->obound = ipc->ibound + ipc->ibsize;

    ipc->sotype = SOCK_STREAM;

    return ipc;
}

//
// listen on a tcp port for client connections
//
long  __stdcall tcp_service(long didx, char *lo_addr_str, f_callup tready)
{
    int      e;
    long     sid;
    ipccb_t *ipc;

    if( 0 == lo_addr_str ) return -1290;

    // don't call tcp_session because we don't need buffers
    ipc = tcp_context(didx, 0, 0);

    if( ipc == 0 ) return -1300;

    sid = ipc->sessid;

    assert( sid > 0 );

    ipc->socket = socket(AF_INET, ipc->sotype, 0);

    ipc->tready = tready;  // connection handler

    // set passive mode before socket events hookup
    ipc->sflags |= SOC_FLAG_PASSIVE;

    // socket events hookup
    ipc_hookup_( get_domain_(didx), ipc );

    set_soaddr( lo_addr_str, &ipc->loaddr );

	// set reuse addres
    e = 1;

    setsockopt( ipc->socket, SOL_SOCKET, SO_REUSEADDR, (char*)&e, sizeof(e));

    e = bind( ipc->socket, (struct sockaddr*)&ipc->loaddr, sizeof(ipc->loaddr) );

    if( e < 0 ) return -1310;

    // put socket to listen
    e = listen( ipc->socket, 3 );

    return sid;
}


//
// create an ipc session context and open a tcp socket 
//
long  __stdcall tcp_session( long didx )
{
    long     sid;
    ipccb_t *ipc;

    ipc = tcp_context(didx, -1, -1);  //default buffer sizes

    if( ipc )
    {
        ipc->socket = socket(AF_INET, ipc->sotype, 0);

        if( ipc->socket == INVALID_SOCKET )
        {
            ipc_print(" tcp_session(): error=%d\n", sock_error());
        }

        sid = ipc->sessid;

        // socket events hookup
        ipc_hookup_( ipc_domain_(sid), ipc );
    }
    else
    {
        sid = -1350; // memory
    }

    return sid;
}


///////////////////////////////////////////////////////////////////////////////

//
// activate a timer, can be calling from outside thread
//
long  __stdcall ipc_loadtmr(long sid, utimer_t *tmrctx)
{
    ipccb_t      *ipc = ipc_context(sid);
    ipc_domain_t *domain;

    assert( tmrctx );

    if( ipc == 0 ) return (-1420);

    // check thread id
    domain = ipc_domain_( sid );

    if( domain == 0 ) return (-1430);

    if( domain->thid != thread_id_() )
    {
        long  ret;
        // out-thread call, thunk
        ipc->apcfun    = tmr_insert_;
        ipc->apcarg[0] = tmrctx;
        ipc->apcsem    = (void*) ~0;  // wait for return value
        ret = ipc_waitapc( domain, ipc );

        return ret;
    }

    // in-thread call
    return tmr_insert_( domain->qtmr, tmrctx );
}


//
// per domain user apc calls
//
void  __stdcall usr_signals( ipc_domain_t *domain )
{
    int               k;
	//int               e;

	for( k = 0; k < domain->ncbs; k++ )
	{
		ipccb_t *ipc = domain->ipcs[k];

		if( ipc == 0 ) continue;

        // check internal async procedure calls
        if( ipc->aflags )
        {
            // an example
            if( ipc->aflags & APC_FLAG_DOCLOSE )
            {
                ipc_closing( ipc ); 
                ipc->aflags ^= APC_FLAG_DOCLOSE;  // clear
            }
        }

		// check if user set any async procedure calls
        if( ipc->apcfun )
        {
            if( ipc->apcfun == ipc_shutoff )
            {
                assert( ipc->sessid );
                ipc->apcval = ipc_shutoff( ipc->sessid );
            }
            else if( ipc->apcfun == ipc_writing )
            {
                ipc->apcval = ipc_writing(ipc->sessid, ipc->apcarg[0], (int)ipc->apcarg[1]);
            }
            else if( ipc->apcfun == tmr_insert_ )
            {
                ipc->apcval = tmr_insert_(domain->qtmr, ipc->apcarg[0]);
            }

            ipc->apcfun = 0; // must clear to prevent double entry

            // tell caller if asked to
            if( ipc->apcsem ) set_event_( ipc->apcsem );
        }
	}
}

///////////////////////////////////////////////////////////////////////////////


//
// spawn a ipc context to accept a client connection
//
long  __stdcall ipc_accept_( ipccb_t *svc )
{
    int      slen;
    ipccb_t *ipc;

    if( svc->sotype != SOCK_STREAM )
    {
        // don't know how to handle
        ipc_print(" ipc_accept_() on non-tcp socket!\n");
        return 0;
    }

    ipc = tcp_context( 0, -1, -1 );  // this is an in-domain call

    if( ipc == 0 )
    {
        return  -1330;
    }

    memcpy(ipc->nuance, svc->nuance, sizeof(svc->nuance));  // signiture

    ipc->sflags |= SOC_FLAG_SPAWNED;

    ipc->peeker = svc->peeker;
    ipc->reader = svc->reader;
    ipc->tready = svc->tready;  // incoming connection handler
    ipc->writer = svc->writer;

    slen = sizeof(ipc->toaddr);

    ipc->socket = accept(svc->socket, (struct sockaddr*)&ipc->toaddr, &slen);

    if( INVALID_SOCKET == ipc->socket )
    {
        // failed - silently discard ipc
        ipc->states = IPC_STAT_INVALID;

        return -1340;  // invalid sid
    }

    return ipc->sessid;
}

//
// called in-thread
//
long  __stdcall ipc_recving( ipccb_t *ipc )
{
    long len = 0;

    if( ipc->peeker )
    {
         // len < 0 if socket error; = 0 graceful shutdown
        len = ipc->peeker( ipc ); 
    }
    else
    {
        // assume we never got blocked

        if( ipc->sotype == SOCK_DGRAM && !(ipc->states & IPC_STAT_CONNECT) )
        {
            int slen = sizeof( ipc->toaddr );

            len = recvfrom( ipc->socket,
                            ipc->ibound + ipc->ipoint,
                            ipc->ibsize - ipc->ipoint,
                            0, 
                            (struct sockaddr*)&ipc->toaddr, 
                            &slen );
        }
        else if( ipc->sotype == SOCK_STREAM || (ipc->states & IPC_STAT_CONNECT) )
        {
            len = recv( ipc->socket, 
                        ipc->ibound + ipc->ipoint,
                        ipc->ibsize - ipc->ipoint,
                        0 );
        }
        else
        {
            ipc_print(" ipc_recving() unknown recv type!\n");
        }

        if( len >= 0 )  // len==0 a graceful shutdown
        {
            if( ipc->reader )
            {
                ipc->ipoint = ipc->reader( ipc, len );
            }
        }
    }
    
	// note: len==0 means a graceful shutdown
    if( len == 0 ) len = -1;

    // negative return will close the session
    return len;
}

//
// for udp sockets do we know incomplete sent?
//
long  __stdcall ipc_sending( ipccb_t *ipc )
{
    long len = 0;

    if( ipc->ocount > 0 )
    {
        if( ipc->states & IPC_STAT_CONNECT )
        {
            len = send(ipc->socket, ipc->obound, ipc->ocount, 0);
        }
        else
        {
            int slen = sizeof( ipc->toaddr );

            len = sendto(ipc->socket, 
                         ipc->obound,
                         ipc->ocount,
                         0,
                         (struct sockaddr*)&ipc->toaddr, 
                         slen );
        }

        if( len < 0 && sock_error()== EWOULDBLOCK ) len = 0;

        if( len > 0 )
        {
            ipc->ocount -= len;

            if( ipc->ocount > 0 )
            {
                memmove( ipc->obound, ipc->obound + len, ipc->ocount );
            }
        }
    }
    else if( ipc->writer )
    {
        len = ipc->writer( ipc );
    }

    // negative return will close the session
    return len;
}

//
// a socket is closed inadvertently
//
void  __stdcall ipc_closing( ipccb_t *ipc )
{
    long  e = 0;

    // 1. close socket
    if( ipc->socket != INVALID_SOCKET )
    {
        sio_close(  ipc->socket );

        ipc->socket = INVALID_SOCKET;
    }

    // 2. clear states
    if( ipc->states & IPC_STAT_CONNECT )
    {
        ipc->states ^= IPC_STAT_CONNECT;
    }

    // 3. clear peer address for passive udp socket
    if( ipc->sflags & SOC_FLAG_PASSIVE )
    {
        memset( &(ipc->toaddr), 0, sizeof(ipc->toaddr) );
    }

    // 4. if user set close routine
    if( ipc->closed )
    {
        e = ipc->closed( ipc );
    }

    // clear buffer
    ipc->ipoint = ipc->ocount = 0;

    // recycle
    if( e == 0 )
    {
        ipc->states = IPC_STAT_INVALID;
    }
}

